使用 Redis 实现分布式锁

背景

在一般的分布式应用中，要安全有效地同步多服务器多进程之间的共享资源访问，就要涉及到分布式锁。目前项目是基于 Tornado 实现的分布式部署，同时也使用了 Redis 作为缓存。参考了一些资料并结合项目自身的要求后，决定直接使用Redis实现全局的分布式锁。

使用 Redis 实现分布式锁

使用 Redis 实现分布式锁最简单方式是创建一对 key-value 值，key 被创建为有一定的生存期，因此它最终会被释放。而当客户端想要释放时，则直接删除 key 。基于不同的 Redis 命令，有两种实现方式：

Redis 官方早期给的一个实现，使用 SETNX，将 value 设置为超时时间，由代码实现锁超时的检测[有缺陷，有限制，并发不高时可用]；
有同学自己的实现：使用 INCR + EXPIRE，利用 Redis 的超时机制控制锁的生存期[不建议使用]；
Redis 官方给的一个改进实现：使用 SET resource-name anystring NX EX max-lock-time（Redis 2.6.12 后支持）实现，利用 Redis 的超时机制控制锁的生存期[Redis 2.6.12 以后建议使用]。

使用 SETNX 实现

Redis 官方最早在 SETNX 命令页给了一个基于该命令的分布式锁实现。

1	Acquire lock: SETNX lock.foo <current Unix time + lock timeout + 1>

1	Release lock: DEL lock.foo

如果 SETNX 返回 1，则表明客户端获取锁成功， lock.foo 被设置为有效 Unix time。客户端操作完成后调用 DEL 命令释放锁。
如果 SETNX 返回 0，则表明锁已经被其他客户端持有。这时我们可以先返回或进行重试等对方完成或等待锁超时。

处理死锁问题：
上述算法中，如果持有锁的客户端发生故障、意外崩溃、或者其他因素因素导致没有释放锁，该怎么解决？。我们可以通过锁的键对应的时间戳来判断这种情况是否发生了，如果当前的时间已经大于lock.foo的值，说明该锁已失效，可以被重新使用。
发生这种情况时，可不能简单的通过DEL来删除锁，然后再SETNX一次，当多个客户端检测到锁超时后都会尝试去释放它，这里就可能出现一个竞态条件：

C1 和 C2 读取 lock.foo 检查时间戳，先后发现超时了。
C1 发送DEL lock.foo。
C1 发送SETNX lock.foo 并且成功了。
C2 发送DEL lock.foo
C2 发送SETNX lock.foo 并且成功了。
ERROR: 由于竞态的问题，C1 和 C2 都获取了锁，这下子问题大了。

幸运的是，使用下面的算法可以避免这个问题。我们看看客户端 C4 是怎么做的：

C4 发送 SETNX lock.foo 想要获取锁。
但是由于发生故障的客户端 C3 仍然持有锁，所以返回 0 给 C4。
C4 发送 GET lock.foo 来检查锁是否过期，如果没超时，则等待或重试。

反之，如果已经超时， C4 则尝试执行下面的命令来获取锁:

1	Acquire lock when time expired: GETSET lock.foo <current Unix timestamp + lock timeout + 1>

通过 GETSET ，C4 拿到的时间戳如果仍然是超时的，那就表明 C4 如愿以偿拿到锁了。
如果在 C4 之前，有个叫 C5 的客户端比 C4 快一步执行了上面的操作，那么 C4 拿到的时间戳是个未超时的值，这时，C4 没有如期获得锁，需要再次等待或重试。留意一下，尽管 C4 没拿到锁，但它改写了 C5 设置的锁的超时值，但是这点微小的误差（一般情况下锁的持有的时间非常短，所以在该竞态下出现的误差是可以容忍的）是可以容忍的。（Note that even if C4 set the key a bit a few seconds in the future this is not a problem）。

为了这个锁的算法更健壮一些，持有锁的客户端在解锁之前应该再检查一次自己的锁是没有超时，再去做 DEL 操作，因为客户端失败的原因很复杂，不仅仅是崩溃也可能是因为某个耗时的操作而挂起，操作完的时候锁因为超时已经锁已经被别人获得，这时就不必解锁了。

仔细的分析这个方案，我们就会发现这里有一个漏洞：Release lock 使用的 DEL 命令不支持 CAS 删除（check-and-set，delete if current value equals old value），在高并发情况下就会有一些问题：确认持有的锁没有超时后执行 DEL 释放锁，由于竞态的存在 Redis 服务器执行命令时锁可能已过期（ “真的” 刚好过期或者被其他客户端竞争锁时设置了一个较小的过期时间而导致过期）且被其他客户端持有。这种情况下将会（非法）释放其他客户端持有的锁。

解决方案： 先确定锁没有超时，再通过 EVAL 命令（在 Redis 2.6 及以上版本提供）在执行 Lua 脚本：先执行 GET 指令获取锁的时间戳，确认和自己的时间戳一致后再执行 DEL 释放锁。

设计缺陷：

上述解决方案并不完美，只解决了过期锁的释放问题，但是由于这个方案本身的缺陷，客户端获取锁时发生竞争（C4 改写 C5 时间戳的例子），那么 lock.foo 的 “时间戳” 将与本地的不一致，这个时候不会执行 DEL 命令，而是等待锁失效，这在高并发的环境下是低效的。
考虑多服务器环境下，需要服务器进行时间同步校准。

在我们的项目中使用了 tornadoredis 库，这个库实现的分布式锁便采用了上述算法。但是在释放锁时有些限制，不过并发量不高的情况下不会有太大的问题，详细的分析参考下述代码注释。实现代码如下所示：

class Lock(object):
    """
    A shared, distributed Lock that uses a Redis server to hold its state.
    This Lock can be shared across processes and/or machines. It works
    asynchronously and plays nice with the Tornado IOLoop.
    """

    LOCK_FOREVER = float(2 ** 31 + 1)  # 1 past max unix time

    def __init__(self, redis_client, lock_name, lock_ttl=None, polling_interval=0.1):
        """
        Create a new Lock object using the Redis key ``lock_name`` for
        state, that behaves like a threading.Lock.
        This method is synchronous, and returns immediately. It doesn't acquire the
        Lock or in fact trigger any sort of communications with the Redis server.
        This must be done using the Lock object itself.
        If specified, ``lock_ttl`` indicates the maximum life time for the lock.
        If none is specified, it will remain locked until release() is called.
        ``polling_interval`` indicates the time between acquire attempts (polling)
        when the lock is in blocking mode and another client is currently
        holding the lock.
        Note: If using ``lock_ttl``, you should make sure all the hosts
        that are running clients have their time synchronized with a network
        time service like ntp.
        """
        self.redis_client = redis_client
        self.lock_name = lock_name
        self.acquired_until = None
        self.lock_ttl = lock_ttl
        self.polling_interval = polling_interval
        if self.lock_ttl and self.polling_interval > self.lock_ttl:
            raise LockError("'polling_interval' must be less than 'lock_ttl'")

    @gen.engine
    def acquire(self, blocking=True, callback=None):
        """
        Acquire the lock.
        Returns True once the lock is acquired.
        If ``blocking`` is False, always return immediately. If the lock
        was acquired, return True, otherwise return False.
        Otherwise, block until the lock is acquired (or an error occurs).
        If ``callback`` is supplied, it is called with the result.
        """

        # Loop until we have a conclusive result
        while 1:

            # Get the current time
            unixtime = int(mod_time.time())

            # If the lock has a limited lifetime, create a timeout value
            if self.lock_ttl:
                timeout_at = unixtime + self.lock_ttl
            # Otherwise, set the timeout value at forever (dangerous)
            else:
                timeout_at = Lock.LOCK_FOREVER
            timeout_at = float(timeout_at)

            # Try and get the lock, setting the timeout value in the appropriate key,
            # but only if a previous value does not exist in Redis
            result = yield gen.Task(self.redis_client.setnx, self.lock_name, timeout_at)

            # If we managed to get the lock
            if result:

                # We successfully acquired the lock!
                self.acquired_until = timeout_at
                if callback:
                    callback(True)
                return

            # We didn't get the lock, another value is already there
            # Check to see if the current lock timeout value has already expired
            result = yield gen.Task(self.redis_client.get, self.lock_name)
            existing = float(result or 1)

            # Has it expired?
            if existing < unixtime:

                # The previous lock is expired. We attempt to overwrite it, getting the current value
                # in the server, just in case someone tried to get the lock at the same time
                result = yield gen.Task(self.redis_client.getset,
                                        self.lock_name,
                                        timeout_at)
                existing = float(result or 1)

                # If the value we read is older than our own current timestamp, we managed to get the
                # lock with no issues - the timeout has indeed expired
                if existing < unixtime:

                    # We successfully acquired the lock!
                    self.acquired_until = timeout_at
                    if callback:
                        callback(True)
                    return

                # However, if we got here, then the value read from the Redis server is newer than
                # our own current timestamp - meaning someone already got the lock before us.
                # We failed getting the lock.

            # If we are not signalled to block
            if not blocking:

                # We failed acquiring the lock...
                if callback:
                    callback(False)
                return

            # Otherwise, we "sleep" for an amount of time equal to the polling interval, after which
            # we will try getting the lock again.
            yield gen.Task(self.redis_client._io_loop.add_timeout,
                           self.redis_client._io_loop.time() + self.polling_interval)

    @gen.engine
    def release(self, callback=None):
        """
        Releases the already acquired lock.
        If ``callback`` is supplied, it is called with True when finished.
        """

        if self.acquired_until is None:
            raise ValueError("Cannot release an unlocked lock")

        # Get the current lock value
        result = yield gen.Task(self.redis_client.get, self.lock_name)
        existing = float(result or 1)

        # 从上下文代码中可以看出，在这个实现中，有一个限制：获取锁的时候设置的 lock_ttl 必须能够保证释放锁时，锁未过期。
        # 否则，当前锁过期后，将会非法释放其他客户端持有的锁。如果无法估计持有锁后代码的执行时间，则可以增加当前锁的过期检测，
        # 当 self.acquired_until <= int(mod_time.time()) 时不执行 DEL 命令。不过，这个限制在一般的应用中倒是可以满足，
        # 所以这个实现不会有太大的问题。
        # 由于 GET、DEL 之间的时间差，以及 DEL 命令发出到 执行 之间的时间差，高并发情况下，锁过期释放的问题依然存在，这个是
        # 算法缺陷。并发不大的情况下，问题不大。
        #
        # 注：这个条件判断 existing >= self.acquired_until 是有这样一个潜在的前提，使用锁的客户端代码正常运行的情况下，
        # 考虑到并发代码使用相同的 lock_ttl 获取锁，竞争失败的客户端将会把锁的过期时间设置的更长一些，这里的判断是有意义的。
        # If the lock time is in the future, delete the lock
        if existing >= self.acquired_until:
            yield gen.Task(self.redis_client.delete, self.lock_name)
        self.acquired_until = None

        # That is it.
        if callback:
            callback(True)

使用 INCR + EXPIRE 实现

该方案的实现来源这篇 blog 《Redis实现分布式全局锁》

客户端A通过 INCR locker.foo 获取名为 locker.foo 的锁，若获取的值为1，则表示获取成功，转入下一步，否则获取失败；
执行 EXPIRE locker.foo seconds 设置锁的过期时间，设置成功转入下一步；
执行共享资源访问；
执行 DEL locker.foo 释放锁。

伪代码如下所示：

if(INCR('locker.foo') == 1)
{
     // 设置锁的超时时间为1分钟，这个可以设置为一个较大的值来避免锁提前过期释放。
     EXPIRE(60)

     // 执行共享资源访问
	 DO_SOMETHING()

     // 释放锁
     DEL('locker.foo')}
}

该实现有一个严重的“死锁”问题：如果 INCR 命令获取锁成功后，EXPIRE 失败，会导致锁无法正常释放。可用的解决方案是：借助 EVAL 命令，将 INCR 、 EXPIRE 操作封装在一个 Lua 脚本中执行，先执行 INCR 命令，成功获取锁后再执行 EXPIRE。以下是示例 Lua 代码：

-- Set a lock
-- KEYS[1]   - key
-- KEYS[2]   - ttl in ms

local key     = KEYS[1]
local ttl     = KEYS[2]

local lockSet = redis.call('incr', key)

if lockSet == 1 then
  redis.call('pexpire', key, ttl)
end

return lockSet

注：由于 EVAL 命令仅在 Redis 2.6 版本后提供，对于之前的版本只能通过 MULTI/EXEC 将 INCR 、 EXPIRE 封装在一个事务中来处理。但是由于 MULTI/EXEC 的限制，没有办法和使用 Lua 脚本一样根据 INCR 执行结果来执行 EXPIRE ，所以如果获取锁失败，会导致 TTL 不断被延长，在高并发的环境里如果拿到锁的进程意外挂掉而没有正常释放锁，锁便只能等到过期才能被其他客户端持有，而这个过期时间的长短取决于获取锁时的竞争激烈情况。该解决方案有严重缺陷，不适合高并发环境。

++实际上，由于不能通过一个原语完成获取锁和设置锁过期时间的操作，即使通过上述 Lua 脚本来获取锁，仍然是有问题的。由于 Redis 事务的特点，只保证 INCR 、 EXPIRE 两条命令在 Redis 上是连续执行的，但当 EXPIRE 命令失败后并不会回滚 INCR 命令，所以 “死锁” 问题依然没有解决（取决于 Redis 的稳定性）。同时，也存在锁过期后非法释放其他客户端持有的锁的问题，且由于依赖 redis 的自动过期机制，便无法检测到此问题。++

使用 SET resource-name anystring NX EX max-lock-time 实现

该方案在 Redis 官方 SET 命令页有详细介绍。
在介绍该分布式锁设计之前，我们先来看一下在从 Redis 2.6.12 开始 SET 提供的新特性，命令 SET key value [EX seconds] [PX milliseconds] [NX|XX]，其中：

EX seconds – 以秒为单位设置 key 的过期时间；
PX milliseconds – 以毫秒为单位设置 key 的过期时间；
NX – 将key 的值设为value ，当且仅当key 不存在，等效于 SETNX。
XX – 将key 的值设为value ，当且仅当key 存在，等效于 SETEX。

注：由于 SET 已经能够取代 SETNX, SETEX, PSETEX 命令，所以在未来的版本中，官方将逐渐放弃这3个命令，并最终移除。

使用 SET 的新特性，改进旧版的分布式锁设计，主要有两个优化：

客户端通过 SET 命令可以同时完成获取锁和设置锁的过期时间：SET lock.foo token NX EX max-lock-time（原子操作，没有INCR 、 EXPIRE两个操作的事务问题），锁将在超时后自动过期，不担心之前设计的 “死锁” 问题，也没有多服务器时间同步校准的问题。
使用 Lua 脚本实现 CAS 删除，使锁更健壮。获取锁时为锁设置一个 token （一个无法猜测的随机字符串），释放锁时先比较 token 的值以保证只释放持有的有效锁。释放锁的 Lua 代码示例：
1
2
3
4
5
6
if redis.call("get",KEYS[1]) == ARGV[1]
then
return redis.call("del",KEYS[1])
else
return 0
end

单点问题

上述实现都有一个单点问题： Redis 节点挂了肿么办？这是个很麻烦的问题，并且由于 Redis 主从复制是异步的，我们便不可能简单地实现互斥锁在节点间的安全迁移。当然一般的项目不会有这么高的要求，就目前我们的项目而言，本身Redis已经是单点。。。

对于这个单点问题，Redis 上有一篇文章提供了一个算法来解决，但是实现比较复杂：

《Distributed locks with Redis（原文）》 / 《使用 Redis 实现分布式锁（译文）》